package com.lrj.datastream.sink;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

/**
 * @author lrj
 * @date 2022/3/29 18:34
 */
public class TestKafkaSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        Properties props = new Properties();
        String bootstrapServer = parameterTool.get("bootstrapServer", "localhost:9092");
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
        String consumerGroupId = parameterTool.get("groupId", "consumer");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("clicks", new SimpleStringSchema(), props);
        String topicId = parameterTool.get("topic", "producer");
        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(bootstrapServer, topicId, new SimpleStringSchema());
        DataStreamSource<String> kafkaStream = env.addSource(consumer);
        kafkaStream.map(s -> "hello:" + s).addSink(producer);
        env.execute(TestKafkaSink.class.getName());
    }
}
